package org.cfk.asyncdownloadtutorial.handler;


import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.cfk.asyncdownloadtutorial.constant.Const;
import org.cfk.asyncdownloadtutorial.dto.AsyncDownloadJobDTO;
import org.cfk.asyncdownloadtutorial.service.AsyncDownloadJobService;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.lang.reflect.Method;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@RequiredArgsConstructor
public class AsyncDownloadJobHandler {

    private final AsyncDownloadJobService asyncDownloadJobService;
    private final RedisTemplate<String, String> redisTemplate;
    private final ApplicationContext applicationContext;
    private final AsyncDownloadJobHandlerContent asyncDownloadJobHandlerContent;

    @SneakyThrows
    @KafkaListener(topics = Const.KAFKA_TOPIC_ASYNC_DOWNLOAD_JOB, concurrency = "1")
    public void asyncJobHandler(String params) {
        log.info("异步下载任务开始处理，参数：{}", params);
        AsyncDownloadJobDTO asyncDownloadJob = asyncDownloadJobService.getById(Long.valueOf(params));
        if (asyncDownloadJob == null) {
            log.error("异步下载任务不存在，任务ID：{}", params);
            return;
        }

        Path tempFile = Files.createFile(Paths.get(asyncDownloadJob.getFileTitle()));
        File file = tempFile.toFile();
        String lockKey = Const.ASYNC_LOCK_KEY + asyncDownloadJob.getId();

        Boolean addLock = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 60, TimeUnit.MINUTES);

        if (addLock) {
            try {
                boolean downloadGenerating = asyncDownloadJobService.setDownloadGenerating(asyncDownloadJob.getId());
                if (!downloadGenerating) {
                    log.warn("设置任务状态为处理中失败，任务ID：{}", asyncDownloadJob.getId());
                    return;
                }

                execute(asyncDownloadJob.getWorker(), asyncDownloadJob.getRequestContent(), file);


                //根据实际情况上传至对应的存储服务,以下为示例代码，请根据实际情况修改
                // 创建本地存储目录
                Path storageDir = Paths.get("storage", "downloads");
                Files.createDirectories(storageDir);
                
                // 生成唯一的文件名
                String fileName = System.currentTimeMillis() + "_" + asyncDownloadJob.getFileTitle();
                Path targetPath = storageDir.resolve(fileName);
                
                // 将临时文件移动到存储目录
                Files.move(file.toPath(), targetPath);
                
                // 更新任务状态为成功，并保存文件路径
                asyncDownloadJobService.setDownloadSuccess(asyncDownloadJob.getId(), targetPath.toString());
                


            } catch (Exception e) {
                log.error("异步下载任务失败：{};{}", e, e.getMessage());
                asyncDownloadJobService.setDownloadFail(asyncDownloadJob.getId());
            } finally {
                if (Objects.nonNull(file)) {
                    Files.deleteIfExists(file.toPath());
                }
                redisTemplate.delete(lockKey);
            }
        } else {
            log.info("异步任务加锁失败：{}", params);
        }
    }

    @SneakyThrows
    private void execute(String worker, String content, File file) {
        Method method = asyncDownloadJobHandlerContent.getByName(worker);
        Class<?>[] parameterTypes = method.getParameterTypes();
        Object[] args = new Object[parameterTypes.length];
        boolean hasOutputStream = false;

        for (int i = 0; i < parameterTypes.length; i++) {
            if (File.class.isAssignableFrom(parameterTypes[i])) {
                args[i] = file;
                hasOutputStream = true;
            } else {
                ObjectMapper objectMapper = new ObjectMapper();
                args[i] = objectMapper.readValue(content, parameterTypes[i]);
            }
        }

        if (!hasOutputStream) {
            throw new RuntimeException("异步下载任务方法参数中没有File类型");
        }

        Object targetBean = applicationContext.getBean(method.getDeclaringClass());
        method.invoke(targetBean, args);
    }
}